package tinyiotserver

import (
	"context"
	"log"
	"sync"
	"sync/atomic"
	"time"

	"gitee.com/mqyqingkong/tiny-iot-server/common"
)

// QueueBroadcastor 将sourceQueue里的数据，广播给多个接收者。例如：
// sourceQueue --msgs(goroutine0)--|---msgs--->  [splitQueue1]  --msgs(goroutine1)-->  broadcast to Target1
//                                 |---msgs--->  [splitQueue2]  --msgs(goroutine2)-->  broadcast to Target2
//                                 |---msgs--->  [splitQueue3]  --msgs(goroutine3)-->  broadcast to Target3
type QueueBroadcastor struct {
	sourceQueue       Queue
	topic             string
	ctx               context.Context
	heartbeatDuration time.Duration //心跳包间隔
	lock              sync.RWMutex
	splitQueues       []Queue

	broadcastNum       int32 //当前Broadcast() goroutine数量
	startSplitQueueNum int32 //当前startSplitQueue() goroutine数量

	splitDone     chan struct{} // startSplitQueue() goroutine是否结束
	broadcastDone chan struct{} // Broadcast() goroutine是否结束
}

func NewQueueBroadcastor(ctx context.Context, sourceQueue Queue, topic string, heartbeatDuration time.Duration) *QueueBroadcastor {
	qb := &QueueBroadcastor{
		sourceQueue:        sourceQueue,
		ctx:                ctx,
		topic:              topic,
		heartbeatDuration:  heartbeatDuration,
		lock:               sync.RWMutex{},
		splitQueues:        []Queue{},
		broadcastNum:       0,
		startSplitQueueNum: 0,
		splitDone:          make(chan struct{}),
		broadcastDone:      make(chan struct{}),
	}
	close(qb.splitDone) // 默认不启动startSplitQueue
	return qb
}

// Broadcast 将sourceQueue里的数据，异步广播给broadcastToTarget。当stopCondition通道关闭时，停止广播。
func (qb *QueueBroadcastor) Broadcast(broadcastToTarget func(msg *Message) error, stopCondition func() <-chan struct{}) error {
	// 为当前Broadcast启动监听
	currentBroadcastNum := atomic.AddInt32(&qb.broadcastNum, 1)
	log.Printf("Broadcast[%d] add for topic: %s \n", currentBroadcastNum-1, qb.topic)
	go func() {
		queue := NewSyncBlockingQueue(NewListQueue(qb.sourceQueue.Size()))
		qb.addSplitQueue(queue)
		defer func() {
			qb.removeSplitQueue(queue)
			currentBroadcastNum := atomic.AddInt32(&qb.broadcastNum, -1)
			if currentBroadcastNum == 0 {
				close(qb.broadcastDone)
			}
			log.Printf("Broadcast[%d] finished for topic: %s \n", currentBroadcastNum, qb.topic)
		}()
		for {
			select {
			case <-qb.ctx.Done():
				return
			case <-stopCondition():
				return
			case <-qb.splitDone: //startSplitQueue()停止时，启动它。
				//防止多个goroutine同时启动startSplitQueue()
				currentStartSplitQueueNum := atomic.AddInt32(&qb.startSplitQueueNum, 1)
				if currentStartSplitQueueNum == 1 {
					select {
					case <-qb.broadcastDone:
						qb.broadcastDone = make(chan struct{})
					default:
					}
					go qb.startSplitQueue()
				} else {
					//防止多个goroutine在startSplitQueue()启动前，死循环
					time.Sleep(100 * time.Millisecond)
				}
			case <-queue.Wait():
				replyMsg, found := queue.PopBack()
				if found {
					err := broadcastToTarget(replyMsg)
					if err != nil {
						return
					}
				}
			}
		}
	}()

	return nil
}

// startSplitQueue 将sourceQueue里的数据，异步推送给临时Queue
func (qb *QueueBroadcastor) startSplitQueue() {
	if qb.heartbeatDuration > 0 {
		qb.startSplitQueueWithHeartbeat()
	} else {
		qb.startSplitQueueWithoutHeartbeat()
	}
}
func (qb *QueueBroadcastor) startSplitQueueWithHeartbeat() {
	qb.splitDone = make(chan struct{})
	heartbeatTimer := time.NewTimer(qb.heartbeatDuration)
	defer func() {
		heartbeatTimer.Stop()
		atomic.StoreInt32(&qb.startSplitQueueNum, 0)
		close(qb.splitDone)
		log.Printf("split topic queue finished for topic: %s \n", qb.topic)
	}()

	for {
		select {
		case <-qb.ctx.Done():
			return
		case <-qb.broadcastDone:
			return
		case <-qb.sourceQueue.Wait():
			qs := qb.getSplitQueues()
			if len(qs) == 0 {
				return
			}
			replyMsg, found := qb.sourceQueue.PopBack()
			if found {
				if !heartbeatTimer.Stop() {
					<-heartbeatTimer.C
				}
				pushFrontToQueues(replyMsg, qs...)
				heartbeatTimer.Reset(qb.heartbeatDuration)
			}
		case <-heartbeatTimer.C:
			qs := qb.getSplitQueues()
			if len(qs) == 0 {
				return
			}
			replyMsg := ResponseMsg(qb.topic, []byte(common.HEARTBEAT))
			log.Printf("heartbeat: %s", replyMsg)
			pushFrontToQueues(replyMsg, qs...)
			heartbeatTimer.Reset(qb.heartbeatDuration)
		}
	}
}

func (qb *QueueBroadcastor) startSplitQueueWithoutHeartbeat() {
	qb.splitDone = make(chan struct{})
	defer func() {
		atomic.StoreInt32(&qb.startSplitQueueNum, 0)
		close(qb.splitDone)
		log.Printf("split topic queue finished for topic: %s \n", qb.topic)
	}()

	for {
		select {
		case <-qb.ctx.Done():
			return
		case <-qb.broadcastDone:
			return
		case <-qb.sourceQueue.Wait():
			qs := qb.getSplitQueues()
			if len(qs) == 0 {
				return
			}
			replyMsg, found := qb.sourceQueue.PopBack()
			if found {
				pushFrontToQueues(replyMsg, qs...)
			}
		}
	}
}

func pushFrontToQueues(msg *Message, queues ...Queue) {
	for _, q := range queues {
		q.PushFront(msg)
	}
}

func (qb *QueueBroadcastor) addSplitQueue(queue Queue) {
	qb.lock.Lock()
	defer qb.lock.Unlock()
	qb.splitQueues = append(qb.splitQueues, queue)
}

func (qb *QueueBroadcastor) removeSplitQueue(queue Queue) {
	qb.lock.Lock()
	defer qb.lock.Unlock()
	newSplitQueues := make([]Queue, 0, len(qb.splitQueues))
	for i := 0; i < len(qb.splitQueues); i++ {
		if (qb.splitQueues[i] == queue) {
			continue
		}
		newSplitQueues = append(newSplitQueues, qb.splitQueues[i])
	}
	qb.splitQueues = newSplitQueues
}

func (qb *QueueBroadcastor) getSplitQueues() []Queue {
	qb.lock.RLock()
	defer qb.lock.RUnlock()
	return qb.splitQueues
}
